package com.hkbigdata.ontimer;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author liuanbo
 * @creat 2024-05-23-9:35
 * @see 2194550857@qq.com
 */
public class Flink04_Event_OnTimer_ReallyWar {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("hadoop102", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                    }
                });

        WatermarkStrategy<WaterSensor> wms = WatermarkStrategy
                .<WaterSensor>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000);

        waterSensorSingleOutputStreamOperator.assignTimestampsAndWatermarks(wms)
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    private Integer lastVc = 0;
                    private Long timerTS = Long.MAX_VALUE;

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        if (value.getVc() > lastVc) {//表示上升
                            if (timerTS == Long.MAX_VALUE) {//表示第一次进来
                                Long timestamp = ctx.timestamp();
                                timerTS = timestamp + 5000L;
                                ctx.timerService().registerEventTimeTimer(timerTS);
                            }
                        } else {
                            ctx.timerService().deleteEventTimeTimer(timerTS);
                            timerTS = Long.MAX_VALUE;
                        }

                        lastVc = value.getVc();
                        System.out.println(lastVc);

                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        System.out.println("定时器触发:" + timestamp / 1000L);

                        System.out.println("定时器被触发......");

                        System.out.println("连续五秒钟传感器的值上升报警" + ctx.getCurrentKey());
                        timerTS = Long.MAX_VALUE;

                    }
                }).print();

        env.execute();

    }
}
